import org.apache.kafka.clients.producer._

import java.util.{Date, Properties}
import scala.collection.mutable

object KafkaUtilTest1 {

  private val props: Properties = new Properties
  private var producer: Producer[String, String] = _

  def init(broker: String): Unit = {
    props.put("bootstrap.servers", broker) // broker 如果有多个,中间使用逗号分隔

    props.put("serializer.class", "kafka.serializer.StringEncoder")
    props.put("request.required.acks", "-1")
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    producer = new KafkaProducer[String, String](props)
  }

  def write2Kafka(topic: String, msg: String): Unit = {
    val producerRecord: ProducerRecord[String, String] = new ProducerRecord[String, String](topic, msg)
    producer.send(producerRecord).get()
    println(System.currentTimeMillis())
    producer.send(producerRecord).get()
    println(System.currentTimeMillis())
    producer.send(producerRecord).get()
    println(System.currentTimeMillis())
    producer.send(producerRecord).get()
    println(System.currentTimeMillis())
    producer.send(producerRecord).get()
    println(System.currentTimeMillis())
    producer.send(producerRecord).get()
    println(System.currentTimeMillis())
    producer.send(producerRecord)
    println(System.currentTimeMillis())
    producer.send(producerRecord)
    println(System.currentTimeMillis())
    producer.send(producerRecord)
    println(System.currentTimeMillis())
    producer.send(producerRecord)
    println(System.currentTimeMillis())
    producer.send(producerRecord)
    println(System.currentTimeMillis())
    producer.send(producerRecord)
    println(System.currentTimeMillis())
  }

  def main(args: Array[String]): Unit = {
    init("dev-cdp-03:9092,dev-cdp-04:9092")
    write2Kafka("test1", "sdafd")
  }


}
